package com.cnpc.bds.sdk.sink.kafka;

import com.cnpc.bds.sdk.sink.kafka.executor.DefaultUpstreamConsumer;
import com.cnpc.bds.sdk.sink.kafka.executor.DownStreamProducer;
import com.cnpc.bds.sdk.sink.kafka.executor.EventBroadcaster;
import com.cnpc.bds.sdk.sink.kafka.handler.Pipeline;
import com.cnpc.bds.sdk.sink.kafka.meta.ConsumerCfg;
import com.cnpc.bds.sdk.sink.kafka.meta.ProducerCfg;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;

/**
 * Sink任务封装构建器.
 */
@Builder
@Slf4j
public class SinkTask {

    /**
     * 上游Kafka消费主题配置.
     */
    private ConsumerCfg upStreamTopic;
    /**
     * 下游生产主题配置.
     */
    private ProducerCfg downStreamTopic;

    /**
     * 消息处理责任链.
     */
    private Pipeline msgHandlePipeline;


    public void start() {
        //消息处理链尽头默认是Kafka生产者.
        msgHandlePipeline.addLast(new DownStreamProducer(downStreamTopic));

        //分配上游消费任务.
        final DefaultUpstreamConsumer upstreamPoller = (DefaultUpstreamConsumer) UpstreamManager.allocate(upStreamTopic);

        //消息处理链注册到一个EventLoop线程，并加入该广播器(broadcaster.join()).
        final EventBroadcaster broadcaster = upstreamPoller.downStreamHook();
        broadcaster.join(msgHandlePipeline);

        log.info("Sink-task start, from [{}] to [{}]!", upStreamTopic.getTopicMeta().getTopic(), downStreamTopic.getTopicMeta().getTopic());
    }

}
